package com.haima.sage.bigdata.api.process.impl;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.haima.sage.bigdata.api.common.Constants;
import com.haima.sage.bigdata.api.common.DbUtils;
import com.haima.sage.bigdata.api.common.EsUtils;
import com.haima.sage.bigdata.api.entity.storage.Storage;
import com.haima.sage.bigdata.api.exception.ValidateException;
import com.haima.sage.bigdata.api.process.ParamProcess;
import com.haima.sage.bigdata.api.vo.BaseParam;
import com.haima.sage.bigdata.api.vo.param.AggParam;
import com.haima.sage.bigdata.api.vo.param.agg.Agg;
import com.haima.sage.bigdata.api.vo.param.agg.AggType;
import com.haima.sage.bigdata.api.vo.param.group.Having;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.http.HttpEntity;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;

public class AggParamProcess extends ParamProcess {
    private static final Logger LOG = LogManager.getLogger(AggParamProcess.class);
    private AggParam param;

    public AggParamProcess(AggParam param) {
        this.param = param;
    }

    protected Object executeES(Storage storage, String table, String pattern, Integer minBatchSize) throws Exception {
        super.executeES(storage, table, pattern, minBatchSize);
        String method = "POST";
        String uri = "/" + table + "/_search";
        Map<String, String> params = Collections.emptyMap();
        String key = storage.getId();
        String entity = genEsJson(table, minBatchSize, storage);

        RestClient client = EsUtils.getClient(key);
        if (client == null) {
            throw new Exception("can't get client");
        }
        Response response = null;
        if (StringUtils.isBlank(entity)) {
            response = client.performRequest(method, uri, params);
        } else {
            HttpEntity httpEntity = new NStringEntity(entity, ContentType.APPLICATION_JSON);
            response = client.performRequest(method, uri, params, httpEntity);
        }

        Integer statusCode = response.getStatusLine().getStatusCode();

        //hits  aggs
        if (statusCode.toString().startsWith("2")) {
            String responseBody = EntityUtils.toString(response.getEntity());
            ObjectMapper jackson = new ObjectMapper();
            JsonNode jsonNode = jackson.readTree(responseBody);
            JsonNode hitsOne = jsonNode.get("aggregations");
            HashMap<String, Object> gbh = new HashMap<String, Object>(4);
            List<HashMap<String, Object>> resList = new ArrayList<HashMap<String, Object>>(10);
            recursESResult(hitsOne,0, gbh, resList);
            if (param.getOffset() >= resList.size()) return Collections.EMPTY_LIST;
            else if (param.getOffset() + param.getLimit() >= resList.size())
                return resList.subList(param.getOffset(), resList.size());
            else
                return resList.subList(param.getOffset(), param.getOffset() + param.getLimit());
        }

        return Collections.EMPTY_LIST;
    }

    //四层循环遍历,处理返回结果
    public Object getEsAggResult(JsonNode jsonNode) {
        List<Map> list = new LinkedList<Map>();
        if (param.getGroupBys() != null && jsonNode.get(param.getGroupBys().get(0)) != null) {
            Iterator iterator = jsonNode.get(param.getGroupBys().get(0)).get("buckets").iterator();
            if (param.getGroupBys().size() > 0) {
                while (iterator.hasNext()) {
                    JsonNode bucket = (JsonNode) iterator.next();
                    Map map = new HashMap();
                    map.put(param.getGroupBys().get(0), bucket.get("key"));
                    if (param.getGroupBys().size() > 1) {
                        JsonNode buckets2 = bucket.get(param.getGroupBys().get(1)).get("buckets");
                        Iterator iterator2 = buckets2.iterator();

                        while (iterator2.hasNext()) {
                            Map map1 = new HashMap(1);
                            JsonNode bucket2 = (JsonNode) iterator2.next();
                            map1.put(param.getGroupBys().get(0), bucket.get("key"));
                            map1.put(param.getGroupBys().get(1), bucket2.get("key"));
                            if (param.getGroupBys().size() > 2) {

                                JsonNode buckets3 = bucket2.get(param.getGroupBys().get(2)).get("buckets");
                                Iterator iterator3 = buckets3.iterator();
                                while (iterator3.hasNext()) {
                                    Map map2 = new HashMap(1);
                                    JsonNode bucket3 = (JsonNode) iterator3.next();
                                    map2.put(param.getGroupBys().get(0), bucket.get("key"));
                                    map2.put(param.getGroupBys().get(1), bucket2.get("key"));
                                    map2.put(param.getGroupBys().get(2), bucket3.get("key"));
                                    if (param.getGroupBys().size() == 4) {

                                        JsonNode buckets4 = bucket3.get(param.getGroupBys().get(3)).get("buckets");
                                        Iterator iterator4 = buckets4.iterator();
                                        while (iterator4.hasNext()) {
                                            Map map3 = new HashMap(1);
                                            JsonNode bucket4 = (JsonNode) iterator4.next();
                                            map3.put(param.getGroupBys().get(0), bucket.get("key"));
                                            map3.put(param.getGroupBys().get(1), bucket2.get("key"));
                                            map3.put(param.getGroupBys().get(2), bucket3.get("key"));
                                            map3.put(param.getGroupBys().get(3), bucket4.get("key"));

                                            //if  第五次,默认支持到第四级聚合
                                            for (Agg iter : param.getAggs()) {

                                                //bucket4.get(iter.getName());
                                                //map3.put(iter.getName(), bucket4.get(iter.getName()).get("value").toString());
                                                String value = bucket4.get(iter.getName()).get("value").toString();
                                                addAgg(iter.getName(), value, map3);
                                            }
                                            list.add(map3);

                                        }

                                    } else {
                                        for (Agg iter : param.getAggs()) {

                                            //bucket3.get(iter.getName());
                                            //map2.put(iter.getName(), bucket3.get(iter.getName()).get("value").toString());
                                            String value = bucket3.get(iter.getName()).get("value").toString();
                                            addAgg(iter.getName(), value, map2);
                                        }
                                        list.add(map2);
                                    }

                                }

                            } else {
                                for (Agg iter : param.getAggs()) {

                                    //bucket2.get(iter.getName());
                                    //map1.put(iter.getName(), bucket2.get(iter.getName()).get("value").toString());
                                    String value = bucket2.get(iter.getName()).get("value").toString();
                                    addAgg(iter.getName(), value, map1);
                                }
                                list.add(map1);
                            }
                        }

                    } else {
                        for (Agg iter : param.getAggs()) {

                            //bucket.get(iter.getName());
                            String value = bucket.get(iter.getName()).get("value").toString();
                            addAgg(iter.getName(), value, map);
                        }
                        list.add(map);
                    }
                }
                /*
                if (list.size() == 1) {
                    return ((LinkedList<Map>) list).getFirst();
                } else {
                    return list;
                }*/
                return list;
            }
        } else {
            Iterator<String> fields = jsonNode.fieldNames();
            List<Map> resultList = new ArrayList();
            Map<String, Object> map = new HashMap<>(1);
            while (fields.hasNext()) {
                String aggName = fields.next();
                String value = jsonNode.get(aggName).get("value").asText();
                addAgg(aggName, value, map);

            }
            resultList.add(map);
            return resultList;

        }

        return null;
    }

    public void recursESResult(JsonNode jnode, int i, HashMap<String, Object> gbh, List<HashMap<String, Object>> resList){ // List<Map<String, Object>> res
        int n = param.getGroupBys() == null ? 0: param.getGroupBys().size();
        if (i == n) {
            HashMap<String, Object> aggh = new HashMap<>(param.getAggs().size());
            for (Agg a: param.getAggs())
                addAgg(a.getName(), jnode.get(a.getName()).get("value").asText(), aggh);
            aggh.putAll(gbh);
            resList.add(aggh);
        } else {
            Iterator<JsonNode> bs = jnode.get(param.getGroupBys().get(i)).get("buckets").elements();
            i++ ;
            while (bs.hasNext()){
                JsonNode jn = bs.next();
                gbh.put(param.getGroupBys().get(i-1), jn.get("key").asText());
                recursESResult(jn, i, gbh, resList);
            }
        }
    }

    private void addAgg(String aggName, String value, Map<String, Object> map) {
        if ("null".equals(value)) {
            map.put(aggName, 0);
        } else if (NumberUtils.isNumber(value)) {
            BigDecimal db = new BigDecimal(value);
            String str = db.toPlainString();
            if (str.matches("[0-9]+")) {
                map.put(aggName, Long.valueOf(str));
            } else {
                String strDouble = db.setScale(2, BigDecimal.ROUND_UP).toPlainString();
                map.put(aggName, Double.valueOf(strDouble));
            }
        } else {
            map.put(aggName, value);
        }
    }

    protected Object executeRedis(Storage storage, String table, Integer maxBatchSize) throws Exception {
        return null;
    }

    @Override
    public void validate() throws ValidateException {
        super.validate((BaseParam) param);
        if (!isValid(param.getGroupBys())) {
            if (isValid(param.getHaving())) {
                throw new ValidateException("if group empty, then having must empty",
                        "如果groupBys为空,那么having也必须为空");
            }
        }
        validate(param.getAggs());
        validate(param.getHaving());
    }

    protected Object executeDB(Storage storage, String table, Integer maxBatchSize) throws Exception {
        String sql = genSQL(table);
        try (Connection conn = DbUtils.getConnection(storage.getId())) {
            return DbUtils.getListResult(sql, conn, maxBatchSize);
//            if (isValid(param.getGroupBys())) {
//               return DbUtils.getListResult(sql, conn, maxBatchSize);
//            } else {
//                return DbUtils.getRowResult(sql, conn);
//            }
        }catch (SQLException e){
            throw new SQLException(e.getMessage());
        }
    }

    /**
     * SELECT ${groupBys}, ${aggs}
     * FROM ${table}
     * WHERE ${where}
     * GROUP BY ${groupBys}
     * HAVING ${havings}
     */
    private String genSQL(String table) throws ValidateException {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT ");
        if (isValid(param.getGroupBys())) {
            for (String item : param.getGroupBys()) {
                sql.append(item).append(",");
            }
        }
        int index = 0;
        for (Agg agg : param.getAggs()) {
            sql.append(agg.getType().getValue()).append("(");
            if (agg.getType() == AggType.CountDistinct) {
                sql.append("DISTINCT ");
            }
            sql.append(agg.getField()).append(") as ");
            if (StringUtils.isEmpty(agg.getName())) {
                sql.append("field" + index);
            } else {
                sql.append(agg.getName());
            }
            sql.append(",");
            index++;
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(" FROM ").append(table);
        // where
        appendWhere(sql, param.getWhere());
        // group by xx having xx
        appendGroupByHaving(sql);
        LOG.info(sql.toString());
        return sql.toString();
    }

    private void appendGroupByHaving(StringBuilder sql) {
        if (param.getGroupBys() != null && param.getGroupBys().size() > 0) {
            sql.append(" GROUP BY ");
            for (String field : param.getGroupBys()) {
                sql.append(field).append(",");
            }
            sql.deleteCharAt(sql.length() - 1);

            if (isValid(param.getHaving())) {
                Having having = param.getHaving();
                sql.append(" HAVING ");
                if(having.getType()!=null) sql.append(having.getType().getValue());
                sql.append("(").append(having.getField()).append(")");
                sql.append(having.getOperator().getValue());
                sql.append(having.getValue());
            }
        }
    }


    private boolean validate(List<Agg> list) throws ValidateException {
        if (list == null || list.size() == 0) {
            throw new ValidateException("aggs must not empty",
                    "aggs不能为空");
        }
        for (Agg agg : list) {
            if (agg.getField() == null) {
                throw new ValidateException("aggs.field must not empty",
                        "aggs.field不能为空");
            }
            if (!Constants.fieldPattern.matcher(agg.getField()).matches()) {
                throw new ValidateException("aggs.field must match regex: " + Constants.fieldPattern.pattern(),
                        "aggs.field必须匹配正则：" + Constants.fieldPattern.pattern());
            }
            if (agg.getType() == null) {
                throw new ValidateException("aggs.type error or must not empty",
                        "aggs.type参数不正确或不能为空");
            }
        }
        return true;
    }

    private boolean validate(Having having) throws ValidateException {
        if (having == null) return true;
        boolean same = false;
        String field = having.getField();
        if (field == null) {
            throw new ValidateException("having.field must not empty",
                    "having.field不能为空");
        }
        if (having.getType() == null) {
            List<Agg> aggs = param.getAggs();
            for (int i = 0; i < aggs.size(); i++) {
                if (field.equals(aggs.get(i).getName())) {
                    same = true;
                    break;
                }
            }
            if (!same)
                throw new ValidateException("Having Field must be Agg Name or field in table when Having Type is not null",
                        "Having Field 必须为Agg Name或者type不为空时表中的字段");
        }


        if (having.getOperator() == null)
            throw new ValidateException("having.operator error or must not empty",
                    "having.operator参数不正确或不能为空");

        return true;
    }

    protected boolean isValid(Having having) {
        if (having == null) {
            return false;
        }
        //if (having.getType() == null) {
        //    return false;
        //}
        if (having.getField() == null) {
            return false;
        }
        if (having.getOperator() == null) {
            return false;
        }
        if (having.getValue() == null) {
            return false;
        }
        return true;
    }


    /**
     * SELECT ${groupBys}, ${aggs}
     * FROM ${table}
     * WHERE ${where}
     * GROUP BY ${groupBys}
     * HAVING ${havings}
     */
    private String genEsJson(String table, Integer maxBatchSize, Storage storage) throws Exception {
        StringBuilder jsonBody = new StringBuilder();
        jsonBody.append("{\"size\": 0,");

        //query where
        if (param.getWhere() != null) {
            jsonBody.append("\"query\":");
            esUtils.appendWhere(jsonBody, param.getTable(), storage, param.getWhere());
        }

        esUtils.appendGroupby(jsonBody, param, storage);
        //System.out.println(jsonBody.length());
        jsonBody.deleteCharAt(jsonBody.length() - 1);
        jsonBody.append("}");
        LOG.info(jsonBody.toString());
        return jsonBody.toString();
    }

}
